package com.changgou.canal.listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.changgou.canal.utils.CanalUtil;
import com.xpand.starter.canal.annotation.CanalEventListener;
import com.xpand.starter.canal.annotation.ListenPoint;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.Map;

@CanalEventListener
public class BusinessListener {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @ListenPoint(schema = "changgou_business",table = "tb_ad")
    public void adUpdate(CanalEntry.EntryType entryType,CanalEntry.RowData rowData){
        System.out.println("广告数据发生变化");
        /*rowData.getBeforeColumnsList().forEach((c)->{
            System.out.println("更改前数据:"+c.getName()+"***"+c.getValue());
        });
        rowData.getAfterColumnsList().forEach((c)->{
            System.out.println("更改后数据:"+c.getName()+"***"+c.getValue());
        });*/


        //使用工具类
        Map<String, String> beforeMap = CanalUtil.convertToMap(rowData.getBeforeColumnsList());
        System.out.println("修改前的数据:"+ JSON.toJSONString(beforeMap));
        Map<String, String> afterMap = CanalUtil.convertToMap(rowData.getAfterColumnsList());
        System.out.println("修改后的数据:"+ JSON.toJSONString(afterMap));

        //向rabbitMQ中发送修改前的信息
        String beforePosition = beforeMap.get("position");
        /*if (beforePosition!=null){
            System.out.println("beforePosition发送消息到MQ");
            rabbitTemplate.convertAndSend("","ad_update_queue",beforePosition);
        }
*/
        //向rabbitMQ中发送修改后的信息
        String afterPosition = afterMap.get("position");
        if (afterPosition!=null && !afterPosition.equals( beforePosition)){
            System.out.println("afterPosition发送消息到MQ");
            rabbitTemplate.convertAndSend("","ad_update_queue",afterPosition);
        }
    }
}
